package com.atguigu.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.bean.TradeSkuOrderBean;
import com.atguigu.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimerService;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * ClassName: DwsTradeSkuOrderWindow
 * Package: com.atguigu.gmall.realtime.app.dws
 * Description:
 *
 * @Author ChenJun(有志男青年)
 * @Create 2023/5/9 11:05
 * @Version 1.0
 */
public class Dws09_TradeSkuOrderWindow_fun04 {
    public static void main(String[] args) {

        //TODO 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // TODO 配置环境信息生产环境得有

//         env.enableCheckpointing(3000L, CheckpointingMode.EXACTLY_ONCE);
//        env.getCheckpointConfig().setCheckpointTimeout(60 * 1000L);
//        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L);
//        env.getCheckpointConfig().enableExternalizedCheckpoints(
//                CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION
//        );
//        env.setRestartStrategy(RestartStrategies.failureRateRestart(
//                10, Time.of(1L, TimeUnit.DAYS), Time.of(3L, TimeUnit.MINUTES)
//        ));
//        env.setStateBackend(new HashMapStateBackend());
//        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");
//        System.setProperty("HADOOP_USER_NAME", "atguigu");

        //TODO 2.从kafka 读取 dwd层下单明细主题数据
        DataStreamSource<String> kafkaDS = env.fromSource(MyKafkaUtil.getKafkaSource("dwd_trade_order_detail",
                "dws09_trade_sku_order_window"), WatermarkStrategy.noWatermarks(), "kafka-source");

        //TODO 3.过滤&转换为JSON对象
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaDS.flatMap(new RichFlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String value, Collector<JSONObject> out) throws Exception {

                if (value != null) {
                    out.collect(JSON.parseObject(value));
                }
            }
        });

        //TODO 4.提取时间生成watermark
        SingleOutputStreamOperator<JSONObject> jsonObjDSWithWMDS = jsonObjDS.assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(2)).withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
            @Override
            public long extractTimestamp(JSONObject element, long recordTimestamp) {
                return element.getLong("create_time");
            }
        }));

        //TODO 5.按照订单明细ID分组
        KeyedStream<JSONObject, String> keyedByDetailIdDS = jsonObjDSWithWMDS.keyBy(json -> json.getString("id"));

        //TODO 6.去重,并转换为JavaBean对象
        SingleOutputStreamOperator<TradeSkuOrderBean> tradeSkuOrderDS = keyedByDetailIdDS.process(new KeyedProcessFunction<String, JSONObject, TradeSkuOrderBean>() {
            private ValueState<JSONObject> valueState;

            @Override
            public void open(Configuration parameters) throws Exception {
                valueState = getRuntimeContext().getState(new ValueStateDescriptor<JSONObject>("value-state", JSONObject.class));
            }

            @Override
            public void processElement(JSONObject value, KeyedProcessFunction<String, JSONObject, TradeSkuOrderBean>.Context ctx, Collector<TradeSkuOrderBean> out) throws Exception {

                //第一条数据到达,则直接保存数据到状态,同时注册5秒定时器
                JSONObject state = valueState.value();
                if (state == null) {
                    valueState.update(value);
                    TimerService timerService = ctx.timerService();
                    timerService.registerEventTimeTimer(timerService.currentWatermark() + 5000L);
                } else {
                    Long lastDts = state.getLong("ts");
                    Long curDts = value.getLong("ts");

                    if (curDts > lastDts) {
                        valueState.update(value);
                    }
                }
            }

            @Override
            public void onTimer(long timestamp, KeyedProcessFunction<String, JSONObject, TradeSkuOrderBean>.OnTimerContext ctx, Collector<TradeSkuOrderBean> out) throws Exception {
                JSONObject value = valueState.value();

                out.collect(TradeSkuOrderBean.builder()
                        .skuId(value.getString("sku_id"))
                        .skuName(value.getString("sku_name"))
                        .originalAmount(value.getBigDecimal("split_original_amount"))
                        .orderAmount(value.getBigDecimal("split_total_amount"))
                        .activityAmount(value.getBigDecimal("split_activity_amount"))
                        .couponAmount(value.getBigDecimal("split_coupon_amount"))
                        .build());

                value.clear();
            }
        });

        //TODO 7.开窗  聚合

        //TODO 8.关联维表,补充维度信息

        //TODO 9.写出数据

        //TODO 10.启动任务

    }
}
